package com.hitqz.robot.biz.websocket.endpoint;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.core.net.NetUtil;
import cn.hutool.core.util.HashUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.crypto.digest.DigestUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.google.common.cache.Cache;
import jakarta.annotation.PreDestroy;
import jakarta.websocket.*;
import jakarta.websocket.server.ServerEndpoint;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.poi.ss.formula.functions.T;
import org.springframework.boot.autoconfigure.web.ServerProperties;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

@ServerEndpoint(value = "/websocket/live")
@Component
@Slf4j
public class LiveEndpoint {
    // 存储sourceKey和会话的对应关系
    private static final Map<String, CopyOnWriteArraySet<Session>> sourceKeySessionMap = new ConcurrentHashMap<>();
    // 用于存储所有打开的WebSocket sessions
    private static final CopyOnWriteArraySet<Session> sessions = new CopyOnWriteArraySet<>();
    // 用于存储视频源地址
    private static final CopyOnWriteArraySet<String> sourcesKeySet = new CopyOnWriteArraySet<>();

    // 全局维护每个 Session 的发送状态
    private static final ConcurrentMap<Session, AtomicBoolean> sessionSendingFlags = new ConcurrentHashMap<>();


    @OnOpen
    public void onOpen(Session session) throws IOException {
        sessions.add(session);
        log.debug("[ws][live]session-id:{} connected.current size :{}", session.getId(), sessions.size());
        String queryString = session.getQueryString();
        String[] pairs = queryString.split("&");
        Map<String, String> queryParams = new HashMap<>();
        for (String pair : pairs) {
            // 使用=分割键和值
            int idx = pair.indexOf("=");
            // 将键值对添加到Map中
            queryParams.put(pair.substring(0, idx), pair.substring(idx + 1));
        }
        String source = queryParams.get("source");
        //大华的rtsp特殊处理
        String subtype = queryParams.get("subtype");
        String channel = queryParams.get("channel");
        if (StrUtil.isNotBlank(subtype)){
            source = source + "&subtype=" + subtype;
        }
        if (StrUtil.isNotBlank(channel)){
            source = source + "&channel=" + channel;
        }
        //默认1080p 1920x1080
        String rs = Optional.ofNullable(queryParams.get("rs")).orElse("1920X1080");
        String sourceKey = DigestUtil.md5Hex(source + rs);
        //保存 sourceKey和session的对应关系 一对多
        if (sourceKeySessionMap.getOrDefault(sourceKey, new CopyOnWriteArraySet<>()).contains(session)) {
            session.close(new CloseReason(CloseReason.CloseCodes.VIOLATED_POLICY, "Duplicate connection"));
            log.warn("[ws][live]sourceKey:{},session-id:{},duplicate connection", sourceKey, session.getId());
        }
        sourceKeySessionMap.computeIfAbsent(sourceKey, k -> new CopyOnWriteArraySet<>()).add(session);
        synchronized (sourceKey.intern()) {
            if (sourcesKeySet.add(sourceKey)) {
                // 开始推流
                // 开启视频转码线程
                ThreadPoolExecutor videoPushThreadPool = SpringUtil.getBean("videoPushThreadPool");
                ServerProperties serverProperties = SpringUtil.getBean(ServerProperties.class);
                log.info("[ws][live]sourceKey:{},source:{}，start push", sourceKey, source);
                videoPushThreadPool.submit(new VideoTranscodeThread(source, rs, videoPushThreadPool, serverProperties));
            }
        }
    }

    @OnMessage
    public void onMessage(Session session, String message) {
        log.info("[ws][live]receive session-id:{},message:{}", session.getId(), message);
    }

    @OnClose
    public void onClose(Session session) {
        // 移除session
        sessions.remove(session);
        log.info("[ws][live]session-id:{},disconnected.current size :{}", session.getId(), sessions.size());
        for (Set<Session> sessions : sourceKeySessionMap.values()) {
            sessions.remove(session);
        }
        sourceKeySessionMap.entrySet().removeIf(entry -> entry.getValue().isEmpty());
    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("[ws][live]session-id:{},error:{}", session.getId(), error.getMessage());
    }

    public static void sendVideoData(String sourceKey, ByteBuffer buffer) {
        byte[] videoData = new byte[buffer.remaining()];  // 创建一个目标数组
        buffer.get(videoData);
        //获取会话
        CopyOnWriteArraySet<Session> sessions = sourceKeySessionMap.get(sourceKey);
        if (CollUtil.isEmpty(sessions)) {
            return;
        }
        Cache<String, Boolean> videoStreamOutputAlive = SpringUtil.getBean("videoStreamOutputAlive");
        for (Session session : sessions) {
            // 获取或创建 AtomicBoolean（线程安全）
            AtomicBoolean sendingFlag = sessionSendingFlags.computeIfAbsent(
                    session,
                    k -> new AtomicBoolean(false)
            );
            // 发送状态  发送完成->发送中
            if (sendingFlag.compareAndSet(false, true)) {
                if (!session.isOpen()) {
                    sessionSendingFlags.remove(session);
                    continue;
                }
                ByteBuffer binary = ByteBuffer.wrap(videoData);
                try {
                    if (session.isOpen()) {
                        session.getBasicRemote().sendBinary(binary);
                    }
                } catch (Exception e) {
                    log.info("[ws][live] send data to {} error ",session.getId());
                }finally {
                    //更新状态 发送完成
                    sessionSendingFlags.get(session).set(false);
                    //流输出状态 有输出
                    videoStreamOutputAlive.put(sourceKey,true);
                }
            }
        }
    }

    @PreDestroy
    public void onDestroy() {
        log.info("Closing all WebSocket connections");
        for (Session session : sessions) {
            try {
                if (session.isOpen()) {
                    session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "Application is shutting down"));
                }
            } catch (IOException e) {
                log.error("Error closing WebSocket session: " + session.getId(), e);
            }
        }
    }




    // 视频转码线程类
    public static class VideoTranscodeThread implements Runnable {

        private final String source;
        private final String rs;
        private final AtomicLong pid = new AtomicLong(-1);

        private final ThreadPoolExecutor videoPushThreadPool;
        private final ServerProperties serverProperties;

        public VideoTranscodeThread(String source, String rs, ThreadPoolExecutor videoPushThreadPool, ServerProperties serverProperties) {
            this.source = source;
            this.rs = rs;
            this.videoPushThreadPool = videoPushThreadPool;
            this.serverProperties = serverProperties;
        }


        @SneakyThrows
        @Override
        public void run() {
            String ip = source.split("@")[1].split(":")[0];
            String port = source.split("@")[1].split(":")[1].split("/")[0];
            String channel = source.split("/")[source.split("/").length - 1];
            //无限循环推流
            while (true) {
                try {
                    while (!ping(ip, 1000)) {
                        log.info("[pushThread] ip:{},port:{},channel:{} ping false", ip, port, channel);
                        Thread.sleep(5000);
                    }
                    log.info("[pushThread] ip:{},port:{},channel:{} ping true", ip, port, channel);
                    // 推流
                    transcodeAndStream(source, rs);
                } catch (Exception e) {
                    if (pid.get() != -1) {
                        ProcessHandle.of(pid.get()).ifPresent(ProcessHandle::destroyForcibly);  // 强制销毁已有进程
                        pid.set(-1);
                    }
                    log.error("[pushThread]ip:{},port:{},channel:{},transcodeAndStream error and restart...,{}", ip, port, channel, e.getMessage());
                }
                Thread.sleep(3000);  // 3秒后重试
            }
        }

        public void transcodeAndStream(String source, String rs) throws IOException, InterruptedException {
            String sourceKey = DigestUtil.md5Hex(source + rs);
            String output = "http://127.0.0.1:" + serverProperties.getPort() + "/admin/rtsp/ws?sourceKey=" + sourceKey;
            //-f rtsp -rtsp_transport tcp告诉ffmpeg要处理的是rtsp流（因为端口映射不使用554端口的情况下ffmpeg可能识别不出来）
            String command = "ffmpeg -f rtsp -rtsp_transport tcp -i " + source + " -c:v mpeg1video -f mpegts -s " + rs + " " + output;
            //启动进程
            ProcessBuilder processBuilder = new ProcessBuilder(command.split(" "));
            Process process = processBuilder.start();
            // 存储进程 ID
            pid.set(process.pid());
            log.info("[pushThread]sourceKey:{},ffmpeg process start,pid:{},command:{}",sourceKey, pid, command);
            // 处理标准输出流
            videoPushThreadPool.submit(() -> {
                try {
                    Thread.sleep(1000);
                    BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
                    String line;
                    while ((line = reader.readLine()) != null && pid.get() != -1) {
                        //log.info("[pushThread] ffmpeg output: {}" , line);
                    }
                    log.warn("[pushThread]sourceKey:{},ffmpeg info log thread exit", sourceKey);
                } catch (Exception e) {
                    log.error("[pushThread]sourceKey:{},ffmpeg info log thread error,{}",sourceKey, e.getMessage());
                }
            });
            // 处理错误输出流
            videoPushThreadPool.submit(() -> {
                try {
                    Thread.sleep(1000);
                    BufferedReader reader = new BufferedReader(new InputStreamReader(process.getErrorStream()));
                    String line;
                    // 即使没有在推流这个输出流仍然会输出数据，所以不能当做推流是否停止的标志
                    while ((line = reader.readLine()) != null && pid.get() != -1) {
                        //log.error("[pushThread] ffmpeg error: {}", line);
                    }
                    log.warn("[pushThread]sourceKey:{},ffmpeg error log thread exit",sourceKey);
                } catch (Exception e) {
                    log.error("[pushThread]sourceKey:{},ffmpeg error log thread error,{}",sourceKey, e.getMessage() );
                }
            });
            videoPushThreadPool.submit(() -> {
                Cache<String, Boolean> videoStreamOutputAlive = SpringUtil.getBean("videoStreamOutputAlive");
                //延迟10秒，等待ffmpeg启动完成
                for (int i = 0; i < 10; i++) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    if (pid.get() == -1){
                        break;
                    }
                }
                while (pid.get() != -1) {
                    try {
                        // 每3秒检测一次
                        for (int i = 0; i < 3; i++) {
                            Thread.sleep(1000);
                            if (pid.get() == -1){
                                break;
                            }
                        }
                        Boolean streamAliveFlag = Optional.ofNullable(videoStreamOutputAlive.getIfPresent(sourceKey)).orElse(false);
                        //log.info("[pushThread]sourceKey:{},ffmpeg check output stream {}",sourceKey,streamAliveFlag? "alive" : "dead");
                        if (!streamAliveFlag) {
                            log.warn("[pushThread]sourceKey:{},ffmpeg check output no data,stop process", sourceKey);
                            ProcessHandle.of(pid.get()).ifPresent(ProcessHandle::destroyForcibly);  // 强制销毁已有进程
                            pid.set(-1);
                            break;
                        }
                    } catch (Exception e) {
                        log.error("[pushThread]sourceKey:{},ffmpeg check output thread error",sourceKey, e);
                    }
                }
                log.warn("[pushThread]sourceKey:{},ffmpeg check output thread exit", sourceKey);
            });
            // 打印线程池使用情况
            log.info("[pushThread]video push thread pool summary,active:{},core:{},max:{}", videoPushThreadPool.getActiveCount(),videoPushThreadPool.getCorePoolSize(),videoPushThreadPool.getMaximumPoolSize());
            // 等待 ffmpeg 进程结束
            int i = process.waitFor();
            log.warn("[pushThread]sourceKey:{},ffmpeg process exit,code:{}", sourceKey, i);
        }


        public static boolean ping(String ip, int timeout) {
            try {
                // 获取当前操作系统
                String os = System.getProperty("os.name").toLowerCase();

                String command = "";

                // Windows 系统
                if (os.contains("win")) {
                    command = "ping -n 1 -w " + timeout + " " + ip;  // Windows
                }
                // macOS 或 Linux 系统
                else if (os.contains("nix") || os.contains("nux") || os.contains("mac")) {
                    command = "ping -c 1 -t " + (timeout / 1000) + " " + ip;  // macOS/Linux
                }

                // 执行命令
                Process process = Runtime.getRuntime().exec(command);
                int exitCode = process.waitFor();
                return (exitCode == 0);
            } catch (Exception e) {
                return false;
            }
        }


    }



}
